package io.kiki.sba.registry.server.metadata.remoting.handler;

import io.kiki.sba.registry.api.Channel;
import io.kiki.sba.registry.common.model.Node;
import io.kiki.sba.registry.common.model.metaserver.inter.heartbeat.HeartbeatRequest;
import io.kiki.sba.registry.common.model.metaserver.nodes.CentralControlNode;
import io.kiki.sba.registry.common.model.slot.SlotBasicInfo;
import io.kiki.sba.registry.common.model.slot.SlotConfig;
import io.kiki.sba.registry.server.metadata.monitor.heartbeat.HeartbeatListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class DefaultHeartbeatListener implements HeartbeatListener<Node> {

    public static final String KEY_TIMESTAMP_GAP_THRESHOLD = "timestamp.gap.threshold";
    private static final Logger logger = LoggerFactory.getLogger(DefaultHeartbeatListener.class);
    private static final long timeGapThreshold = Long.getLong(KEY_TIMESTAMP_GAP_THRESHOLD, 2000);

    private final String dataCenter;

    private final Channel channel;

    private volatile boolean isValidChannel = true;


    public DefaultHeartbeatListener(String dataCenter, Channel channel) {
        this.dataCenter = dataCenter;
        this.channel = channel;
    }


    @Override
    public void onHeartbeat(HeartbeatRequest<Node> heartbeatRequest) {
        checkIfDataCenterMatched(heartbeatRequest);
        checkIfTimeSynced(heartbeatRequest);
        checkIfSlotBasicInfoMatched(heartbeatRequest);
        closeIfChannelNotValid();
    }

    private void closeIfChannelNotValid() {
        if (!isValidChannel) {

            channel.close();
        }
    }

    private void checkIfTimeSynced(HeartbeatRequest<Node> heartbeat) {
        long timestamp = heartbeat.getTimestamp();
        if (System.currentTimeMillis() - timestamp > timeGapThreshold) {
            logger.error("[checkIfTimeSynced] {} timestamp[{}] is far behind mine[{}]", heartbeat.getNode(), timestamp, System.currentTimeMillis());
        }
    }

    private void checkIfSlotBasicInfoMatched(HeartbeatRequest<Node> heartbeat) {

        if (heartbeat.getNode() instanceof CentralControlNode) {
            return;
        }

        SlotBasicInfo slotBasicInfo = heartbeat.getSlotBasicInfo();
        if (!SlotConfig.Slot_Function.equals(slotBasicInfo.getSlotFunc())) {
            logger.error("[checkIfSlotBasicInfoMatched] {} slot function not match(meta-server: [{}], receive: [{}]", heartbeat.getNode(), SlotConfig.Slot_Function, slotBasicInfo.getSlotFunc());
            isValidChannel = false;
        }
        if (SlotConfig.SLOT_NUM != slotBasicInfo.getSlotNum()) {
            logger.error("[checkIfSlotBasicInfoMatched] {} slot number not match(meta-server: [{}], receive: [{}]", heartbeat.getNode(), SlotConfig.SLOT_NUM, slotBasicInfo.getSlotNum());
            isValidChannel = false;
        }
        if (SlotConfig.SLOT_REPLICAS != slotBasicInfo.getSlotReplicas()) {
            logger.error("[checkIfSlotBasicInfoMatched] {} slot replicas not match(meta-server: [{}], receive: [{}]", heartbeat.getNode(), SlotConfig.SLOT_REPLICAS, slotBasicInfo.getSlotReplicas());
            isValidChannel = false;
        }
    }

    private void checkIfDataCenterMatched(HeartbeatRequest<Node> heartbeat) {
        String dc = heartbeat.getDataCenter();
        if (!this.dataCenter.equalsIgnoreCase(dc)) {
            logger.error("[checkIfDataCenterMatched] {} datacenter not match(meta-server: [{}], node: [{}]", heartbeat.getNode(), this.dataCenter, dc);
            isValidChannel = false;
        }
    }
}
